package com.yanggu.flink.tableapi_sql.define_table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.row
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings}

/**
 * 1. 创建流式执行环境
 * 2. 创建表执行环境
 * 3. 定义表
 * 4. 在表执行环境中注册表
 * 5. 执行sql或者table api
 * 6. 执行任务
 */
/**
 * 使用Values的方式定义table
 */
object ValuesForTable {

  def main(args: Array[String]): Unit = {
    //1. 创建流式执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    //2. 创建表的执行环境
    val settings = EnvironmentSettings
      .newInstance()
      //使用流的方式
      .inStreamingMode()
      .build()

    val tableEnvironment = StreamTableEnvironment.create(environment, settings)

    val keywords = "hello flink hello blink hello muscle hello power"
    val expressions = keywords
      .split(" ")
      //.map(data => row(data, 1))
      //这里只能用Int.box的方式, 进行自动装箱
      .map(data => row(data, Int.box(1)))

    //定义table
    //先定义表结构, 然后定义数据
    val table = tableEnvironment.fromValues(
      DataTypes.ROW(
        DataTypes.FIELD("word", DataTypes.STRING().notNull()),
        DataTypes.FIELD("frequency", DataTypes.INT().notNull())
        //这里的可变参数需要使用: _*的方式
      ), expressions: _*
    )

    //打印table的schema
    table.printSchema()
    //使用tableEnv注册临时视图
    tableEnvironment.createTemporaryView("word_count", table)

    //执行查询
    tableEnvironment.sqlQuery(
      """
        |SELECT
        |  word,
        |  SUM(frequency)
        |FROM
        |  word_count
        |GROUP BY
        |  word
        |""".stripMargin)
      .execute()
      .print()

  }

}
